/*
 * Copyright 2006-2022 ahtelit.com, All Rights Reserved.
 */

package com.telit.firecommand.listener;

import cn.hutool.json.JSONUtil;
import com.telit.firecommand.service.SysUserService;
import com.telit.upms.constants.UpmsConstants;
import com.telit.upms.dto.SysUserOperationStatusDto;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;


/**
 * 统一平台系统用户操作监听
 *
 * @author yuezhuang
 * @since 2023-01-17
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class SysUserOperationListener {

    private final SysUserService sysUserService;

    @KafkaListener(
            topics = "${isolation-prefix.kafka:}" + UpmsConstants.KafkaTopic.USER_DDL_CALL_BACK_TOPIC,
            groupId = "${isolation-prefix.kafka:}${spring.kafka.consumer.group-id}"
    )
    public void listen(String message, Acknowledgment ack) {
        try {
            SysUserOperationStatusDto dto = JSONUtil.toBean(message, SysUserOperationStatusDto.class);
            sysUserService.dealSysUser(dto);
        } catch (Exception e) {
            log.error("用户信息操作失败：", e);
        } finally {
            //手动提交offset
            ack.acknowledge();
        }
    }

}
